#include "my_tcp.h"
#include "my_worker.h"
#include "my_event_loop.h"
#include "comm/my_inet.h"
#include "comm/my_hooks.h"
#include "config/my_cache_conf.h"

#define MAX_DO_ACCEPT_ONCE                  32

const char *tcp_accept_nconns_tag       = "tcp_accept_nconns";
const char *tcp_accept_conns_tag        = "tcp_accept_conns";

my_tcp_listener_t g_tcp_listener_list   = {
    .next = NULL
};

typedef struct {
    int                     nthreads;
    my_event_loop_data_t    *datas;
    my_thread_base_t        *threads;
} my_tcp_accepter_t;

static my_tcp_accepter_t tcp_accepter = {
    0, NULL, NULL
};

typedef struct {
    int                     sockfd;             /* accept fd */
    my_event_handler_t      accept_handler;     /* server handler */
    my_dispatch_event_t     dispatch_event;
} my_tcp_accept_dispatch_event_t;

// run in worker thread.
static void
my_tcp_accept_dispatch_handler(my_dispatch_event_t *dispatch_event)
{
    my_tcp_accept_dispatch_event_t *accept_dispatch_event;
    accept_dispatch_event = dispatch_event->data;

    my_conn_t *conn = my_thread_calloc(sizeof(my_conn_t));
    my_conn_init(conn, MY_CONN_EPOLL_ET);
    conn->fd = accept_dispatch_event->sockfd;
    // save accept_handler for http keepalive.
    conn->server_handler = accept_dispatch_event->accept_handler;
    conn->read.handler = accept_dispatch_event->accept_handler;

    my_thread_free(accept_dispatch_event);

    // notify using read event.
    conn->read.handler(&conn->read);
}

// run in accept thread.
static my_tcp_accept_dispatch_event_t *
my_tcp_accept_dispatch_event_alloc(int sockfd, my_event_handler_t accept_handler)
{
    my_tcp_accept_dispatch_event_t *accept_dispatch_event;
    accept_dispatch_event = my_thread_calloc(sizeof(my_tcp_accept_dispatch_event_t));
    accept_dispatch_event->sockfd = sockfd;
    accept_dispatch_event->accept_handler = accept_handler;
    accept_dispatch_event->dispatch_event.data = accept_dispatch_event;
    accept_dispatch_event->dispatch_event.handler = my_tcp_accept_dispatch_handler;

    return accept_dispatch_event;
}

// run in accept thread.
static my_bool_t
my_tcp_accept_sockfd_set(int accept_fd, my_tcp_listener_t *ls)
{
    int                             opt_val;
    my_syscall_result_t             res;

    if (ls->nodelay) {
        res = my_syscall_setsockopt(accept_fd,
            IPPROTO_TCP, TCP_NODELAY, &opt_val, sizeof(int));
        if (res.eno != 0) {
            MY_LOG_ERROR("setsockopt(TCP_NODELAY) %d failed, errno: %d",
                accept_fd, res.eno);
            return MY_FALSE;
        }
    }

    if (ls->quickack) {
        res = my_syscall_setsockopt(accept_fd,
            IPPROTO_TCP, TCP_QUICKACK, &opt_val, sizeof(int));
        if (res.eno != 0) {
            MY_LOG_ERROR("setsockopt(TCP_QUICKACK) %d failed, errno: %d",
                accept_fd, res.eno);
            return MY_FALSE;
        }
    }

    return MY_TRUE;
} 

// run in accept thread.
static void
my_tcp_accept_handle(my_event_t *event)
{
    my_conn_t                       *c;
    my_tcp_listener_t               *ls;
    my_syscall_result_t             res;
    my_tcp_accept_dispatch_event_t  *accept_dispatch_event;

    c = event->data;
    ls = c->data;

    int accept_fd;
    for (int i = 0; i < MAX_DO_ACCEPT_ONCE; i++) {
        res = my_syscall_accept4(c->fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC);
        if (res.eno != 0) {
            break; // EAGAIN or othread error.
        }

        accept_fd = (int)res.res;
        if (!my_tcp_accept_sockfd_set(accept_fd, ls)) {
            MY_LOG_ERROR("set accept fd %d tcp options failed", accept_fd);
            my_syscall_close(accept_fd);
            continue;
        }

        accept_dispatch_event =
            my_tcp_accept_dispatch_event_alloc(accept_fd, ls->accept_handler);
        my_worker_processor_post_event(&accept_dispatch_event->dispatch_event);
    }
}

// run in accept thread.
static void
my_tcp_accepter_loop_init(my_event_loop_data_t *loop_data)
{
    my_conn_t           *conn, *conns;
    my_tcp_listener_t   *listener;
    int                 i = 0;
    int                 n = 0;

    listener = g_tcp_listener_list.next;
    while (listener != NULL) {
        n++;
        listener = listener->next;
    }

    conns = my_thread_calloc((int64_t)sizeof(my_conn_t) * n);

    listener = g_tcp_listener_list.next;
    for (i = 0; i < n; i++) {
        conn = conns + i;
        my_conn_init(conn, MY_CONN_EPOLL_LT);
        conn->fd = listener->sockfd;
        conn->data = listener;
        conn->read.handler = my_tcp_accept_handle;

        if (!my_event_loop_start_io(conn)) {
            MY_LOG_FATAL("start tcp listener %s io failed", listener->listen_addr);
        }

        if (!my_event_loop_enabled_io(conn, MY_FALSE)) {
            MY_LOG_FATAL("enabled tcp listener %s io failed", listener->listen_addr);
        }

        listener = listener->next;
    }

    my_thread_store_safe_put(tcp_accept_nconns_tag, (void *)(intptr_t)n);
    my_thread_store_safe_put(tcp_accept_conns_tag, conns);

    my_hook_chain_exec(tcp_accept_init_chian, loop_data);
}

// run in accept thread.
static void
my_tcp_accepter_loop_quit(my_event_loop_data_t *loop_data)
{
    int         nconns;
    my_conn_t   *conn, *conns;

    my_hook_chain_exec(tcp_accept_quit_chian, loop_data);

    nconns = (int)(intptr_t)my_thread_store_get(tcp_accept_nconns_tag);
    conns = my_thread_store_get(tcp_accept_conns_tag);

    if (conns == NULL) {
        MY_LOG_FATAL("unreachable");
    }

    // can not be close.
    for (int i = 0; i < nconns; i++) {
        conn = conns + i;
        my_event_loop_remove_conn(conn);
    }

    my_thread_free(conns);
}

// run in main thread.
static my_bool_t
my_tcp_listener_open(my_tcp_listener_t *l)
{
    int port;
    if (!my_address_parser(l->listen_addr, &l->address, &port) || port <= 0) {
        MY_LOG_ERROR("invalid listener addr: %s", l->listen_addr);
        return MY_FALSE;
    }

    my_syscall_result_t res;
    res = my_syscall_socket(l->address.sockaddr.sa_family, SOCK_STREAM, 0);
    if (res.eno != 0) {
        MY_LOG_ERROR("socket() %s failed, errno: %d", l->listen_addr, res.eno);
        return MY_FALSE;
    }

    l->sockfd = (int)res.res;

    int opt_val = 1;

    if (l->reuseport) {
        res = my_syscall_setsockopt(l->sockfd, SOL_SOCKET, SO_REUSEPORT,
            &opt_val, sizeof(int));
        if (res.eno != 0) {
            MY_LOG_ERROR("setsockopt(SO_REUSEPORT) %s failed, errno: %d",
                l->listen_addr, res.eno);
            return MY_FALSE;
        }
    }

    if (l->reuseaddr) {
        res = my_syscall_setsockopt(l->sockfd, SOL_SOCKET, SO_REUSEADDR,
            &opt_val, sizeof(int));
        if (res.eno != 0) {
            MY_LOG_ERROR("setsockopt(SO_REUSEADDR) %s failed, errno: %d",
                l->listen_addr, res.eno);
            return MY_FALSE;
        }
    }

    if (l->ipv6only && l->address.sockaddr.sa_family == AF_INET6) {
        res = my_syscall_setsockopt(l->sockfd, IPPROTO_IPV6, IPV6_V6ONLY,
            &opt_val, sizeof(int));
        if (res.eno != 0) {
            MY_LOG_ERROR("setsockopt(IPV6_V6ONLY) %s failed, errno: %d",
                l->listen_addr, res.eno);
            return MY_FALSE;
        }
    }

    res = my_syscall_set_nonblock(l->sockfd);
    if (res.eno != 0) {
        MY_LOG_ERROR("setnonblock %s failed, errno: %d", l->listen_addr, res.eno);
        return MY_FALSE;
    }

    res = my_syscall_set_close_exec(l->sockfd);
    if (res.eno != 0) {
        MY_LOG_ERROR("setcloseexec %s failed, errno: %d", l->listen_addr, res.eno);
        return MY_FALSE;
    }

    res = my_syscall_bind(l->sockfd, &l->address.sockaddr,
        (socklen_t)my_address_socklen(&l->address));
    if (res.eno != 0) {
        MY_LOG_ERROR("bind %s failed, errno: %d", l->listen_addr, res.eno);
        return MY_FALSE;
    }

    if (l->address.sockaddr.sa_family == AF_UNIX) {
        mode_t mode = (S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH);
        if (chmod(l->address.sockaddr_un.sun_path, mode) == -1) {
            MY_LOG_ERROR("chmod unix: %s failed, errno: %d",
                l->address.sockaddr_un.sun_path, errno);
            return MY_FALSE;
        }
    }

    res = my_syscall_listen(l->sockfd, l->backlog);
    if (res.eno != 0) {
        MY_LOG_ERROR("listen() to %s failed, errno: %d", l->listen_addr, res.eno);
        return MY_FALSE;
    }

    return MY_TRUE;
}

// run in main thread.
static void
my_tcp_listener_close(my_tcp_listener_t *l)
{
    if (l->sockfd == INVALID_FD) {
        MY_LOG_ERROR("close invalid tcp listener");
        return;
    }

    my_syscall_close(l->sockfd);
}

// run in main thread.
void my_tcp_accepter_start()
{
    int nevents;
    int accept_threads;

    my_cache_conf_t *cache_conf = my_cache_conf_get();
    my_service_conf_t *service_conf = &cache_conf->service_conf;

    nevents = (int)my_conf_item_int64_get(&service_conf->nevents);
    accept_threads = (int)my_conf_item_int64_get(&service_conf->accept_threads);

    if (nevents <= 0 || accept_threads <= 0) {
        MY_LOG_FATAL("nevents and accept threads must be greate than 0");
    }

    if (g_tcp_listener_list.next == NULL) {
        MY_LOG_WARN("no tcp listener");
        return;
    }

    my_tcp_listener_t *listener;
    listener = g_tcp_listener_list.next;

    while (listener != NULL) {
        if (!my_tcp_listener_open(listener)) {
            MY_LOG_ERROR("open %s listener failed, exit process", listener->listen_addr);
            exit(1);
        }

        listener = listener->next;
    }

    my_event_loop_data_t        *datas;
    my_thread_base_t            *threads;

    datas = my_thread_calloc((int64_t)sizeof(my_event_loop_data_t) * accept_threads);
    threads = my_thread_calloc((int64_t)sizeof(my_thread_base_t) * accept_threads);

    for (int i = 0; i < accept_threads; i++) {
        my_event_loop_data_init(datas+i,
            my_tcp_accepter_loop_init,
            my_tcp_accepter_loop_quit,
            nevents);
    }

    tcp_accepter.nthreads = accept_threads;
    tcp_accepter.threads = threads;
    tcp_accepter.datas = datas;

    for (int i = 0; i < accept_threads; i++) {
        my_thread_base_start(
            threads+i,
            my_event_loop_runner,
            datas+i,
            0, 0
        );
    }
}

// run in main thread.
void
my_tcp_accepter_stop()
{
    if (tcp_accepter.nthreads == 0) {
        return;
    }

    int                 i;
    my_tcp_listener_t   *listener;

    for (i = 0; i < tcp_accepter.nthreads; i++) {
        my_event_loop_quit(tcp_accepter.threads+i);
    }

    for (i = 0; i < tcp_accepter.nthreads; i++) {
        my_thread_base_join(tcp_accepter.threads+i, NULL);
    }

    listener = g_tcp_listener_list.next;
    while (listener != NULL) {
        my_tcp_listener_close(listener);
        listener = listener->next;
    }

    my_thread_free(tcp_accepter.datas);
    my_thread_free(tcp_accepter.threads);
}

////////////////////////////// tcp connect //////////////////////////////

typedef struct {
    my_tcp_connect_handle_pt        handler;
    void                            *data;
} my_tcp_connect_data_t;

// run in worker threads.
static void
my_tcp_connect_done(my_conn_t *conn, my_result_t res)
{
    my_tcp_connect_handle_pt    handler;
    void                        *data;

    my_tcp_connect_data_t *conn_data = conn->data;
    handler = conn_data->handler;
    data = conn_data->data;
    my_thread_free(conn_data);

    conn->data = NULL;
    conn->write.handler = NULL;

    if (res != MY_SUCCESS) {
        if (conn->fd != INVALID_FD) {
            my_conn_close(conn);
        }
        my_thread_free(conn);
        return;
    }

    handler(data, conn, res);
}

// run in worker threads.
static void
my_tcp_conn_wait_connect_handler(my_event_t *event)
{
    my_conn_t *conn = event->data;
    my_event_loop_remove_conn(conn);

    if (event->timedout) {
        MY_LOG_ERROR("connect timedout, fd: %d", conn->fd);
        event->timedout = 0;
        my_tcp_connect_done(conn, MY_FAILED);
        return;
    }

    int opt_val     = 0;
    int opt_size    = sizeof(opt_val);

    my_syscall_result_t res;
    res = my_syscall_getsockopt(conn->fd, SOL_SOCKET, SO_ERROR, &opt_val, (socklen_t *)&opt_size);
    if (res.eno != 0 || opt_val != 0) {
        MY_LOG_ERROR("connect failed, fd: %d, errno: %d, opt_val: %d",
            conn->fd, res.res, opt_val);
        my_tcp_connect_done(conn, MY_FAILED);
        return;
    }

    my_tcp_connect_done(conn, MY_SUCCESS);
}

// run in worker threads.
static void
my_tcp_connect_setup_helper(my_conn_t *conn, my_address_t *address,
    int64_t timeout_ms, my_tcp_connect_opts_t *opts)
{
    my_thread_base_t *thread_base = current_thread_base();
    if (!thread_base) {
        MY_LOG_FATAL("current thread is not event_loop");
    }

    my_syscall_result_t res;
    res = my_syscall_socket(address->sockaddr.sa_family, SOCK_STREAM, 0);
    if (res.eno != 0) {
        MY_LOG_ERROR("socket() failed, errno: %d", res.eno);
        return my_tcp_connect_done(conn, MY_FAILED);
    }

    conn->fd = (int)res.res;

    int opt_val = 1;

    if (opts->nodelay) {
        res = my_syscall_setsockopt(conn->fd, IPPROTO_TCP, TCP_NODELAY, &opt_val, sizeof(int));
        if (res.eno != 0) {
            MY_LOG_ERROR("setsockopt(TCP_NODELAY) %d failed, errno: %d", conn->fd, res.eno);
            return my_tcp_connect_done(conn, MY_FAILED);
        }
    }

    if (opts->quickack) {
        res = my_syscall_setsockopt(conn->fd, IPPROTO_TCP, TCP_QUICKACK, &opt_val, sizeof(int));
        if (res.eno != 0) {
            MY_LOG_ERROR("setsockopt(TCP_QUICKACK) %d failed, errno: %d", conn->fd, res.eno);
            return my_tcp_connect_done(conn, MY_FAILED);
        }
    }

    res = my_syscall_set_nonblock(conn->fd);
    if (res.eno != 0) {
        MY_LOG_ERROR("setnonblock %d failed, errno: %d", conn->fd, res.eno);
        return my_tcp_connect_done(conn, MY_FAILED);
    }

    res = my_syscall_set_close_exec(conn->fd);
    if (res.eno != 0) {
        MY_LOG_ERROR("setcloseexec %d failed, errno: %d", conn->fd, res.eno);
        return my_tcp_connect_done(conn, MY_FAILED);
    }

    res = my_syscall_connect(conn->fd, &address->sockaddr, (socklen_t)my_address_socklen(address));
    // connect success
    if (res.eno == 0) {
        return my_tcp_connect_done(conn, MY_SUCCESS);
    }

    if (res.eno != EINPROGRESS) {
        MY_LOG_ERROR("connect() failed, fd: %d, errno: 5d", conn->fd, res.eno);
        return my_tcp_connect_done(conn, MY_FAILED);
    }

    conn->write.handler = my_tcp_conn_wait_connect_handler;

    if (timeout_ms > 0) {
        my_event_loop_add_timer(&conn->write, timeout_ms);
    }

    my_event_loop_start_io(conn);
    my_event_loop_enabled_io(conn, MY_TRUE);
    return; // wait event loop callback
}


// run in worker threads.
void
my_tcp_connect_setup(my_address_t *address,
    int64_t timeout_ms, my_tcp_connect_opts_t *opts,
    my_tcp_connect_handle_pt connect_handler, void *data)
{
    my_tcp_connect_data_t *connect_data =
        my_thread_calloc(sizeof(my_tcp_connect_data_t));
    connect_data->handler = connect_handler;
    connect_data->data = data;

    my_conn_t *conn = my_thread_calloc(sizeof(my_conn_t));
    my_conn_init(conn, MY_CONN_EPOLL_ET);
    conn->data = connect_data;

    my_tcp_connect_setup_helper(conn, address, timeout_ms, opts);
}